package com.ruoyi.energy.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.energy.domain.EmsTimeSeriesData;
import com.ruoyi.energy.service.IMqttMessageService;
import com.ruoyi.energy.service.ITDengineTimeSeriesService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * MQTT消息处理服务实现类
 * 
 * @author ruoyi
 * @date 2025-06-29
 */
@Service
public class MqttMessageServiceImpl implements IMqttMessageService {

    private static final Logger log = LoggerFactory.getLogger(MqttMessageServiceImpl.class);

    @Autowired
    private ITDengineTimeSeriesService tdengineTimeSeriesService;

    @Autowired
    private MessageChannel mqttOutboundChannel;

    /**
     * 处理接收到的MQTT消息
     */
    @Override
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        try {
            String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
            String payload = message.getPayload().toString();
            
            log.info("接收到MQTT消息 - 主题: {}, 内容: {}", topic, payload);
            
            // 解析消息并存储到TDengine
            parseAndStoreMessage(topic, payload);
            
        } catch (Exception e) {
            log.error("处理MQTT消息失败", e);
        }
    }

    /**
     * 解析消息并存储到TDengine数据库
     * 支持单个设备数据和多设备数组格式
     */
    private void parseAndStoreMessage(String topic, String payload) {
        try {
            // 首先尝试解析为JSON对象或数组
            Object jsonData = JSON.parse(payload);

            if (jsonData instanceof JSONArray) {
                // 处理数组格式的多设备数据
                parseAndStoreArrayMessage(topic, (JSONArray) jsonData);
            } else if (jsonData instanceof JSONObject) {
                // 处理单个设备数据
                parseSingleDeviceMessage(topic, (JSONObject) jsonData);
            } else {
                log.warn("不支持的JSON格式 - 主题: {}, 内容: {}", topic, payload);
            }

        } catch (Exception e) {
            log.error("解析和存储MQTT消息失败 - 主题: {}, 内容: {}", topic, payload, e);
        }
    }

    /**
     * 处理数组格式的多设备数据
     */
    private void parseAndStoreArrayMessage(String topic, JSONArray jsonArray) {
        try {
            List<EmsTimeSeriesData> timeSeriesDataList = new ArrayList<>();
            int successCount = 0;
            int failCount = 0;

            log.info("接收到数组格式MQTT消息，包含 {} 个设备数据 - 主题: {}", jsonArray.size(), topic);

            // 遍历数组中的每个设备数据
            for (int i = 0; i < jsonArray.size(); i++) {
                try {
                    Object item = jsonArray.get(i);
                    if (item instanceof JSONObject) {
                        JSONObject deviceData = (JSONObject) item;

                        // 从设备数据中获取设备编码，如果没有则从主题中提取
                        String deviceCode = getDeviceCodeFromData(deviceData, topic);
                        if (deviceCode == null) {
                            log.warn("无法获取设备编码，跳过数组索引: {} - 主题: {}", i, topic);
                            failCount++;
                            continue;
                        }

                        // 构建时序数据对象
                        EmsTimeSeriesData timeSeriesData = buildTimeSeriesData(deviceCode, topic, deviceData);
                        timeSeriesDataList.add(timeSeriesData);
                        successCount++;

                    } else {
                        log.warn("数组中的元素不是JSON对象，跳过索引: {} - 主题: {}", i, topic);
                        failCount++;
                    }
                } catch (Exception e) {
                    log.error("处理数组中的设备数据失败，索引: {} - 主题: {}", i, topic, e);
                    failCount++;
                }
            }

            // 批量存储到TDengine数据库
            if (!timeSeriesDataList.isEmpty()) {
                boolean success = tdengineTimeSeriesService.batchInsertTimeSeriesData(timeSeriesDataList);
                if (success) {
                    log.info("批量MQTT消息数据已成功存储到TDengine - 成功: {}, 失败: {}, 主题: {}",
                            successCount, failCount, topic);
                } else {
                    log.error("批量MQTT消息数据存储到TDengine失败 - 成功: {}, 失败: {}, 主题: {}",
                            successCount, failCount, topic);
                }
            } else {
                log.warn("没有有效的设备数据可存储 - 主题: {}", topic);
            }

        } catch (Exception e) {
            log.error("处理数组格式MQTT消息失败 - 主题: {}", topic, e);
        }
    }

    /**
     * 处理单个设备数据
     */
    private void parseSingleDeviceMessage(String topic, JSONObject jsonData) {
        try {
            // 从设备数据中获取设备编码，如果没有则从主题中提取
            String deviceCode = getDeviceCodeFromData(jsonData, topic);
            if (deviceCode == null) {
                log.warn("无法获取设备编码 - 主题: {}", topic);
                return;
            }

            // 构建时序数据对象
            EmsTimeSeriesData timeSeriesData = buildTimeSeriesData(deviceCode, topic, jsonData);

            // 存储到TDengine数据库
            boolean success = tdengineTimeSeriesService.insertTimeSeriesData(timeSeriesData);

            if (success) {
                log.info("MQTT消息数据已成功存储到TDengine - 设备: {}", deviceCode);
            } else {
                log.error("MQTT消息数据存储到TDengine失败 - 设备: {}", deviceCode);
            }

        } catch (Exception e) {
            log.error("处理单个设备MQTT消息失败 - 主题: {}", topic, e);
        }
    }

    /**
     * 从设备数据中获取设备编码，优先从数据中获取，否则从主题中提取
     */
    private String getDeviceCodeFromData(JSONObject deviceData, String topic) {
        // 优先从数据中获取设备编码
        if (deviceData.containsKey("deviceCode")) {
            String deviceCode = deviceData.getString("deviceCode");
            if (deviceCode != null && !deviceCode.trim().isEmpty()) {
                return deviceCode.trim();
            }
        }

        // 如果数据中没有设备编码，则从主题中提取
        return extractDeviceCodeFromTopic(topic);
    }

    /**
     * 转换energyType为标准编码格式
     * 支持多种输入格式：数字字符串、英文名称等
     */
    private String convertEnergyTypeToCode(String energyType) {
        if (energyType == null || energyType.trim().isEmpty()) {
            return "1"; // 默认为电力
        }

        String type = energyType.trim().toLowerCase();

        // 如果已经是数字编码，直接返回
        if (type.matches("^[1-5]$")) {
            return type;
        }

        // 转换英文名称为数字编码
        switch (type) {
            case "electric":
            case "electricity":
            case "power":
            case "电力":
                return "1";
            case "gas":
            case "natural_gas":
            case "燃气":
                return "2";
            case "water":
            case "水":
                return "3";
            case "steam":
            case "蒸汽":
                return "4";
            case "other":
            case "others":
            case "其他":
                return "5";
            default:
                log.warn("未知的energyType类型: {}，使用默认值1(电力)", energyType);
                return "1"; // 默认为电力
        }
    }

    /**
     * 从主题中提取设备编码
     */
    private String extractDeviceCodeFromTopic(String topic) {
        try {
            // 主题格式: ems/device/{deviceCode}/data 或 ems/sensor/{deviceCode}/data
            String[] parts = topic.split("/");
            if (parts.length >= 3) {
                return parts[2]; // 设备编码在第3个位置
            }
        } catch (Exception e) {
            log.error("提取设备编码失败: {}", topic, e);
        }
        return null;
    }

    /**
     * 构建时序数据对象
     */
    private EmsTimeSeriesData buildTimeSeriesData(String deviceCode, String topic, JSONObject jsonData) {
        EmsTimeSeriesData data = new EmsTimeSeriesData();
        
        // 设置基本信息
        data.setDeviceCode(deviceCode);
        data.setTs(new Date()); // 使用当前时间，也可以从消息中获取
        
        // 优先从JSON消息中获取energyType，否则根据主题类型设置
        String energyType = getStringFromJson(jsonData, "energyType");
        if (energyType != null && !energyType.trim().isEmpty()) {
            // 如果消息中包含energyType字段，转换为标准格式
            data.setEnergyType(convertEnergyTypeToCode(energyType.trim()));
        } else {
            // 否则根据主题类型设置默认值（使用数字编码）
            if (topic.contains("/device/")) {
                data.setEnergyType("1"); // 电力设备
            } else if (topic.contains("/sensor/")) {
                data.setEnergyType("1"); // 传感器默认为电力类型
            } else if (topic.contains("/energy/")) {
                data.setEnergyType("1"); // 能耗数据默认为电力类型
            } else if (topic.contains("/gateway/")) {
                data.setEnergyType("1"); // 网关数据默认为电力类型
            } else {
                data.setEnergyType("5"); // 其他类型
            }
        }
        
        // 从JSON中提取数据字段
        if (jsonData.containsKey("timestamp")) {
            data.setTs(new Date(jsonData.getLongValue("timestamp")));
        }
        
        // 电力相关数据
        data.setPower(getBigDecimalFromJson(jsonData, "power"));
        data.setTotalEnergy(getBigDecimalFromJson(jsonData, "totalEnergy"));
        data.setInstantEnergy(getBigDecimalFromJson(jsonData, "instantEnergy"));
        data.setVoltage(getBigDecimalFromJson(jsonData, "voltage"));
        data.setCurrent(getBigDecimalFromJson(jsonData, "current"));
        data.setPowerFactor(getBigDecimalFromJson(jsonData, "powerFactor"));

        // 流量相关数据
        data.setFlowRate(getBigDecimalFromJson(jsonData, "flowRate"));
        data.setTotalFlow(getBigDecimalFromJson(jsonData, "totalFlow"));
        data.setPressure(getBigDecimalFromJson(jsonData, "pressure"));

        // 环境数据
        data.setTemperature(getBigDecimalFromJson(jsonData, "temperature"));
        data.setHumidity(getBigDecimalFromJson(jsonData, "humidity"));
        
        // 状态和质量 - 安全提取字符串
        data.setStatus(getStringFromJson(jsonData, "status"));
        data.setQuality(getStringFromJson(jsonData, "quality"));
        
        // 扩展字段 - 安全提取字符串
        data.setExtend1(getStringFromJson(jsonData, "extend1"));
        data.setExtend2(getStringFromJson(jsonData, "extend2"));
        data.setExtend3(getStringFromJson(jsonData, "extend3"));
        
        return data;
    }

    /**
     * 发送MQTT消息
     */
    @Override
    public void sendMessage(String topic, String payload) {
        sendMessage(topic, payload, 1, false);
    }

    /**
     * 发送MQTT消息
     */
    @Override
    public void sendMessage(String topic, String payload, int qos, boolean retained) {
        try {
            Message<String> message = MessageBuilder.withPayload(payload)
                    .setHeader(MqttHeaders.TOPIC, topic)
                    .setHeader(MqttHeaders.QOS, qos)
                    .setHeader(MqttHeaders.RETAINED, retained)
                    .build();
            
            mqttOutboundChannel.send(message);
            log.info("MQTT消息发送成功 - 主题: {}, 内容: {}", topic, payload);
            
        } catch (Exception e) {
            log.error("MQTT消息发送失败 - 主题: {}, 内容: {}", topic, payload, e);
        }
    }

    /**
     * 订阅新主题
     */
    @Override
    public void subscribe(String topic, int qos) {
        // 这里可以实现动态订阅逻辑
        log.info("订阅MQTT主题: {}, QoS: {}", topic, qos);
    }

    /**
     * 取消订阅主题
     */
    @Override
    public void unsubscribe(String topic) {
        // 这里可以实现取消订阅逻辑
        log.info("取消订阅MQTT主题: {}", topic);
    }

    /**
     * 从JSON对象中获取BigDecimal值
     */
    private BigDecimal getBigDecimalFromJson(JSONObject jsonData, String key) {
        if (jsonData.containsKey(key)) {
            Object value = jsonData.get(key);
            if (value != null) {
                try {
                    return new BigDecimal(value.toString());
                } catch (NumberFormatException e) {
                    log.warn("无法转换字段{}的值为BigDecimal: {}", key, value);
                }
            }
        }
        return null;
    }

    /**
     * 从JSON对象中安全获取字符串值
     */
    private String getStringFromJson(JSONObject jsonData, String key) {
        if (jsonData.containsKey(key)) {
            Object value = jsonData.get(key);
            if (value != null) {
                if (value instanceof byte[]) {
                    try {
                        return new String((byte[]) value, "UTF-8");
                    } catch (Exception e) {
                        log.warn("字节数组转字符串失败，字段: {}", key, e);
                        return "";
                    }
                }
                return value.toString();
            }
        }
        return "";
    }
}
